package com.cotte.flow.kafka;

import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import net.sf.json.JSONArray;
import net.sf.json.JSONObject;

@Component
public class Consumer {
	
	protected final Logger logger = LoggerFactory.getLogger(this.getClass());
	
	
	@KafkaListener(group = "${kafka.consumer.group.id}",topics = {"${kafka.consumer.topic}"})
	public void consumerMes(ConsumerRecord<String, String> record) {
		System.out.println("TESTkafka的value: " + record.value().toString());
		JSONObject obj = JSONObject.fromObject(record.value().toString());
		String jhnr = obj.get("JHNR").toString();
		JSONArray arr = JSONArray.fromObject(jhnr);
		for(int i=0;i<arr.size();i++) {
			JSONObject obj_jhnr = JSONObject.fromObject(arr.get(i));
			String cjdm = obj_jhnr.get("CJDM").toString();
			String scggdh = obj_jhnr.get("SCGGDH").toString();
			String[] dhList = scggdh.split(",");
			for(String dh:dhList) {
				String[] dhArr = dh.split(":");
				String syscode = dhArr[0];
				String cate = dhArr[1];
				System.out.println("syscode:"+syscode+"-cate:"+cate);
			}
			System.out.println("cjdm:"+cjdm);
		}
	}
}
